package cn.rocksolid.sport.common.mq;

import cn.rocksolid.sport.common.Constant.MQ;
import cn.rocksolid.sport.common.Constant.Symbol;
import cn.rocksolid.sport.common.context.RockSolidContextHolder;
import cn.rocksolid.sport.common.context.RockSolidContextImpl;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.ImmutableList;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.destination.DynamicDestinationResolver;

import javax.annotation.Nonnull;
import javax.jms.*;

@Configuration
public class ActiveMQConfiguration {

  @Value("${rocksolid.node-no}")
  private Integer nodeNo;

  @Bean(name = MQ.CONN_FACTORY_NAME)
  public ConnectionFactory activeMQPooledConnectionFactory() {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
    factory.setTrustedPackages(ImmutableList.of("cn.rocksolid"));
    return factory;
  }

  @Bean(name = MQ.JMS_TEMPLATE_NAME)
  public JmsTemplate activeMQJmsTemplate(@Qualifier(MQ.CONN_FACTORY_NAME) ConnectionFactory connectionFactory) {
    JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
    jmsTemplate.setMessageConverter(new FastjsonMessageConverter());
    return jmsTemplate;
  }

  @Bean(name = MQ.SUBSCRIBE_FACTORY)
  public DefaultJmsListenerContainerFactory activeMQTopicListenerContainerFactory(
          @Qualifier(MQ.CONN_FACTORY_NAME) ConnectionFactory connectionFactory) {
    DefaultJmsListenerContainerFactory queueFactory = new DefaultJmsListenerContainerFactory();
    queueFactory.setMessageConverter(new FastjsonMessageConverter());
    queueFactory.setConnectionFactory(connectionFactory);
    queueFactory.setDestinationResolver(new DynamicDestinationResolver() {
      @Override
      @Nonnull
      protected Queue resolveQueue(@Nonnull final Session session, @Nonnull final String queueName)
              throws JMSException {
        return super.resolveQueue(session, String.join(Symbol.DOT, MQ.CP, String.valueOf(nodeNo), queueName));
      }
    });
    return queueFactory;
  }

  private static class FastjsonMessageConverter implements MessageConverter {

    private void attachCtx(Message msg) throws JMSException {
      msg.setStringProperty(MQ.MSG_HEADER_CTX, JSON.toJSONString(RockSolidContextHolder.getContext()));
    }

    private void detachCtx(Message msg) throws JMSException {
      RockSolidContextHolder.setContext(
              JSON.parseObject(msg.getStringProperty(MQ.MSG_HEADER_CTX), RockSolidContextImpl.class));
    }

    @Override
    @Nonnull
    public Message toMessage(@Nonnull final Object obj, @Nonnull final Session session) throws JMSException {
      if (obj instanceof Message) {
        return (Message) obj;
      } else if (obj instanceof String) {
        TextMessage tm = session.createTextMessage((String) obj);
        attachCtx(tm);
        return tm;
      } else {
        StreamMessage sm = session.createStreamMessage();
        attachCtx(sm);
        sm.writeString(JSON.toJSONString(obj));
        return sm;
      }
    }

    @Override
    @Nonnull
    public Object fromMessage(@Nonnull final Message message) throws JMSException {
      if (message instanceof TextMessage) {
        TextMessage tm = (TextMessage) message;
        detachCtx(tm);
        return tm.getText();
      } else if (message instanceof StreamMessage) {
        StreamMessage sm = (StreamMessage) message;
        detachCtx(sm);
        return JSON.parseObject(sm.readString());
      }
      return message;
    }
  }
}
